/**
 * 
 */
package org.cmcc.pcap.packet.udp;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.cmcc.pcap.packet.capture.JCaptures;
import org.cmcc.pcap.packet.util.PacketAnalysisConfig;
import org.pcap4j.packet.IpPacket;
import org.pcap4j.packet.UdpPacket;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Map.Entry;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;


/**
 * @author Amish
 * @date 2018年12月24日 上午11:04:51
 * 
 */
public class UdpDispatcher {
	public static Logger logger = LogManager.getLogger(UdpDispatcher.class);
	JCaptures jCaptures;
	
	int udpConnectionCacheNum = 7;//UdpConnectionCache数量，add(生产者线程数量)和get的工作线程的数量(消费者线程的数量)默认与该值相等.TODO 当为0时考虑采用动态弹性的线程数量管理
	int udpConnectionCacheGetratio = 1;//每个TcpConnectionCache对应的get工作线程数量(消费者线程的数量),即UdpProtocolDispatcher的数量，默认为1，与TcpConnectionCache数量相等
	ConcurrentHashMap<String,DispatcherContext> dispatcherContexts;
	
	boolean statistic = true;//statistic 为true：启动StatisticTask扫描线程处理的实时统计信息   为false：则不启动.StatisticTask如果启动，需要显示调用TcpDispatcher的stop，来结束线程，否则主线程不会退出TODO 不为true主线程扫描完文件就退出了,如何完善
	boolean statisticStart = false;
	AtomicInteger udpPacketNum;
	long startTimeLong = 0;//记录所有包处理开始时间
	long stopTimeLong = 0;//记录所有包处理结束时间
	private ScheduledExecutorService  scheduledExecutorService;
	public boolean isOveroad = false;
	int overloadPacketsNum = 0;
	public Timestamp timestamp;
	String taskId;
	
	public UdpDispatcher(JCaptures jCaptures,String taskId){
		this.jCaptures = jCaptures;
        //TODO 读配置文件，初始化各种参数值
		this.udpPacketNum = new AtomicInteger(0);
		this.dispatcherContexts = new ConcurrentHashMap<String,DispatcherContext>();
		this.udpConnectionCacheNum = PacketAnalysisConfig.getConfigAsInt("udp.connection.cache.num");
		this.udpConnectionCacheGetratio = PacketAnalysisConfig.getConfigAsInt("udp.connection.cache.getratio");
		this.overloadPacketsNum = PacketAnalysisConfig.getConfigAsInt("udp.deal.overload");
		this.timestamp = jCaptures.timestamp;
		this.taskId = taskId;
	}
	
	private synchronized void startStastics() {
		this.startTimeLong = System.currentTimeMillis();//记录所有包处理开始时间
		this.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
		scheduledExecutorService.scheduleAtFixedRate(new StatisticTask(this), 2, 10,TimeUnit.SECONDS);//启动一个定时统计任务
		this.statisticStart = true;
	}
	
	public void dispatcherUdpPacket(IpPacket udpPacket) {
		this.udpPacketNum.incrementAndGet();
		while(this.isOveroad) {// 过负荷，阻塞
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				logger.error(e.getCause(),e);
			}
		}
		
		if(statistic && !statisticStart){//如果scheduledExecutorService已经启动，将不重复启动
			this.startStastics();
		}
		
		String dispatcherKeyStr = this.gengerteDispatcherKeyStr(udpPacket, udpConnectionCacheNum);
		logger.debug("dispatcher keyStr is: "+ dispatcherKeyStr );
		
		DispatcherContext dispatcherContext;
		if(dispatcherContexts.containsKey(dispatcherKeyStr)){
			dispatcherContext = (DispatcherContext) dispatcherContexts.get(dispatcherKeyStr);
		}else{
			dispatcherContext = new DispatcherContext(this.udpConnectionCacheGetratio);//构造方法里会new一个UdpProtocolDispatcher	
			dispatcherContext.startUdpProtocolDispatcher();//生成UDP Packet对应的消费者/工作线程		
			dispatcherContexts.put(dispatcherKeyStr, dispatcherContext);
		}
		
		dispatcherContext.execute(udpPacket);
		
		
	}
	
	private String gengerteDispatcherKeyStr(IpPacket ipPacket,int cacheSize){

		UdpPacket udpPacket = ipPacket.get(UdpPacket.class);

		int src_ip = ipPacket.getHeader().getSrcAddr().hashCode();
		int src_port = udpPacket.getHeader().getSrcPort().value();
		int dst_ip = ipPacket.getHeader().getDstAddr().hashCode();
		int dst_port = udpPacket.getHeader().getDstPort().value();
		
		long unionCode = src_ip+src_port+dst_ip+dst_port;
		int dispatcherKeyCode = (int) (Math.abs(unionCode) % cacheSize);
		return Integer.toString(dispatcherKeyCode);
	}
	
	public class DispatcherContext {
		UdpConnectionCache udpConnectionCache;
		ExecutorService executorService;
		int cacheGetratio = 1;
		ArrayList<Thread> workerThreads;
		
		public DispatcherContext(int cacheGetratio){
			this.cacheGetratio = cacheGetratio;
			this.udpConnectionCache = new UdpConnectionCache();
			//this.executorService = Executors.newSingleThreadExecutor();
			this.executorService = new ThreadPoolExecutor(1, 1,
					0L, TimeUnit.MILLISECONDS,
					new LinkedBlockingQueue<Runnable>());//为了获得事件队列的长度，重写了这部分实现.此处为一个单线程线程池，所有提交的任务执行前缓存在无界队列里，可保证提交的任务串行执行。
			workerThreads = new ArrayList<Thread>();
			for(int i = 0 ; i < this.cacheGetratio ; i++){
				UdpProtocolDispatcher worker = new UdpProtocolDispatcher(this.udpConnectionCache);
				worker.setTaskId(taskId);
				Thread thread = new Thread(worker);
				thread.setDaemon(true);//设置为daemon，在主线程退出时该线程也退出。
				workerThreads.add(thread);
			}
		}
		
		public void execute(IpPacket ipPacket) {
			this.executorService.execute(new UdpProducer(ipPacket,this.udpConnectionCache));
		}

		public void startUdpProtocolDispatcher(){//每个DispatcherContext对应一个，每个线程对应一个UdpHandler
			for( Thread thread : workerThreads){
				thread.start();
			}
		}
		public UdpConnectionCache getUdpConnectionCache(){
			return this.udpConnectionCache;
		}
		public ExecutorService getExecutorService() {
			return this.executorService;
		}
		public void stop() {
			this.executorService.shutdown();
		}
	}
	
	
	class StatisticTask implements Runnable{
		UdpDispatcher udpDispatcher;
		public StatisticTask(UdpDispatcher udpDispatcher) {
			this.udpDispatcher = udpDispatcher;
		}
		@Override
		public void run() {
			try{
				jCaptures.udp_packet_exist = true;
				StringBuffer sb = new StringBuffer();
				int i =1;//仅仅是序号
				int dispatcher_queueSum =0;
				int cache_totalSum = 0;
				int cache_pickupSum = 0;
				int cache_handledSum = 0;
				sb.append("\n##############################################################################\n");
				sb.append("All Statistic Information"+" [taskid : "+taskId+"]\n");
				sb.append("MainThread receive udp Packet : "+udpPacketNum.get()+"\n");
				for(DispatcherContext dispatcherContext : dispatcherContexts.values()){
					ExecutorService es = dispatcherContext.getExecutorService();
					ThreadPoolExecutor tpe = (ThreadPoolExecutor) es;
					int queueSize = tpe.getQueue().size();    //线程池任务队列中未被处理的UDP Packet数量
					dispatcher_queueSum = dispatcher_queueSum + queueSize;
					
					UdpConnectionCache udpConnectionCache = dispatcherContext.getUdpConnectionCache();
					int totalPacketNum = udpConnectionCache.totalPacketNum.get();
					int pickupPacketNum = udpConnectionCache.pickupPacketNum.get();
					int handledPacketNum = udpConnectionCache.handledPacketNum.get();
					
					cache_totalSum = cache_totalSum + totalPacketNum;
					cache_pickupSum = cache_pickupSum + pickupPacketNum;
					cache_handledSum = cache_handledSum + handledPacketNum;
					
					sb.append("[ "+i+" ] "
							 +"Queue : < "+queueSize+" > "
					         +"| CacheTotal : < "+totalPacketNum+" > "
					         +"| CachePickup : < "+cache_pickupSum+" > "
					         +"| CacheHandled : < "+cache_handledSum+" > \n");
								
					i++;
				}
				sb.append("--------------------------------------------------------------------------------\n");
				sb.append("[SUM] "
				         +"Queue : < "+dispatcher_queueSum+" > "
				         +"| CacheTotal : < "+cache_totalSum+" > "
				         +"| CachePickup : < "+cache_pickupSum+" > "
				         +"| CacheHandled : < "+cache_handledSum+" > \n");
				sb.append("\n##############################################################################\n");
				
				if(cache_totalSum == cache_pickupSum && cache_pickupSum == cache_handledSum) {
					sb.append("\n##############################################################################");
					stopTimeLong = System.currentTimeMillis();//记录所有包处理结束时间，误差正负10s，10s为统计任务的周期启动时间
					sb.append("\n  [ all packet deal duration  "+(stopTimeLong-startTimeLong)+"  ms ]");
					sb.append("\n##############################################################################");
					logger.info(sb.toString());
//                    jCaptures.complete = true;
					jCaptures.udp_complete = true;
                    logger.debug("udp packets done !");
					return;
				}
				logger.info(sb.toString());
				
			}catch (Exception e){
				logger.error(e.getCause(),e);
			}
			
		}
	}
	
	public void stop(){
		logger.info("[ stoping TcpDispatcher ]");
		if(this.scheduledExecutorService != null)
		    this.scheduledExecutorService.shutdownNow();
		this.statisticStart = false;
		for(Entry<String, DispatcherContext> entry : dispatcherContexts.entrySet()) {
			entry.getValue().stop();
		}
		dispatcherContexts.clear();
		
		this.udpPacketNum.set(0);
		
		logger.info("[ stoped TcpDispatcher ]");
	}

}
